package com.weishe.loc;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class LocReducer extends Reducer<Text, Loc, Text, Text> {
	public static Calendar c = Calendar.getInstance();
	public static SimpleDateFormat SDF = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

	private Text outValue = new Text();

	@Override
	protected void reduce(Text key, Iterable<Loc> values, Reducer<Text, Loc, Text, Text>.Context context)
			throws IOException, InterruptedException {

		long timeSpan = 0l;
		Map<String, Long> map = new HashMap<String, Long>();
		long first = 0l;
		long next = 0l;
		int i = 1;
		Loc lastLoc = null;

		if (key.toString().equals("0054784943|17")) {
			int kkk = 0;
		}

		for (Loc loc : values) {

			if ((i + 2) / 2 == 1) {
				first = loc.getTime();
				lastLoc = loc;
			} else {
				next = loc.getTime();
				// 获取到第偶数个的时候的时候
				long time = next - first;
				if (time < 3600000l) {
					Long otime = map.get(lastLoc.getLoc());
					if (otime != null) {
						map.put(lastLoc.getLoc(), time + otime.longValue());
					} else {
						map.put(lastLoc.getLoc(), time);
					}
				}
			}
			i++;
		}
		// 当出现最后一个为单数的时候
		if (i / 2 > 0 && i > 1) {
			if (map == null) {
				System.out.println("map null key :" + key);
				return;
			}

			if (lastLoc == null) {
				System.out.println("lastLoc null key :" + key + "map:" + map.size());
				return;
			}
			Long otime = map.get(lastLoc.getLoc());
			long time = 0l;
			switch (lastLoc.getHours()) {
			case 9:
				c.setTimeInMillis(lastLoc.getTime());
				c.set(Calendar.HOUR, 9);
				time = c.getTimeInMillis() - lastLoc.getTime();

				break;
			case 17:
				c.setTimeInMillis(lastLoc.getTime());
				c.set(Calendar.HOUR, 17);
				time = c.getTimeInMillis() - lastLoc.getTime();
				break;
			case 24:
				c.setTimeInMillis(lastLoc.getTime());
				c.set(Calendar.HOUR, 24);
				time = c.getTimeInMillis() - lastLoc.getTime();
				break;
			}
			if (otime != null) {
				if (time < 3600000l) {
					map.put(lastLoc.getLoc(), time + otime.longValue());
				}
			} else {
				if (time < 3600000l) {
					map.put(lastLoc.getLoc(), time);
				}
			}
		}
		// 取时间最长的三个地方
		ThreeMap m = new ThreeMap();
		for (String k : map.keySet()) {
			m.put(k, map.get(k));
		}

		// 输出
		HashMap<String, Long> tm = m.getAll();
		String out = "";
		for (String lc : tm.keySet()) {
			outValue.set(lc + "_" + tm.get(lc));
			context.write(key, outValue);
		}

	}

}
